package com.code.coinprice.handlers.route.sub;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.code.coinprice.consts.CommonConstant;
import com.code.coinprice.handlers.RedisHandler;
import com.code.coinprice.handlers.route.CoinPriceInvokeHandler;
import com.code.coinprice.models.CoinPriceSource;
import com.code.coinprice.models.RelativeCoinPrice;
import com.code.coinprice.services.CoinPriceDictService;
import com.code.coinprice.services.CoinPriceSourceRouteService;
import com.code.coinprice.services.CoinPriceSourceService;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
import org.apache.http.NameValuePair;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Component(CommonConstant.CMC + CommonConstant.COIN_PRICE_INVOKE_BEAN_NAME_SUFFIX)
public class CmcCoinPriceInvokeApiHandler extends CoinPriceInvokeHandler {

    private static final Logger log = LoggerFactory.getLogger(CmcCoinPriceInvokeApiHandler.class);

    /**
     * 核心线程数
     */
    private static final int CORE_THREADS = 20;

    /**
     * CMC币价数据列表的大小上限
     */
    private static final int CMC_RESULT_SIZE_LIMIT = 5000;

    private static ThreadPoolExecutor coinMarketCapThreadPool = new ThreadPoolExecutor(CORE_THREADS, 50, 300, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(50),
            r -> new Thread(r, "coin_market_cap_task_pool_" + r.hashCode()), new ThreadPoolExecutor.DiscardOldestPolicy());

    /**
     * CMC获取币价地址的后缀
     */
    private static final String URL_SUFFIX = "/v1/cryptocurrency/listings/latest";

    @Value("${rest.template.socket-time-out}")
    private int socketTimeOut;

    @Value("${rest.template.connect-time-out}")
    private int connectTimeOut;

    @Value("${rest.template.connect-request-timeout}")
    private int connectRequestTimeout;

    @Resource
    private CoinPriceDictService coinPriceDictService;

    @Resource
    private CoinPriceSourceService coinPriceSourceService;

    @Resource
    private RedisHandler redisHandler;

    @Resource
    private CoinPriceSourceRouteService coinPriceSourceRouteService;

    @Override
    public List<RelativeCoinPrice> getCoinPriceList(CoinPriceSource coinPriceSource) {
        if (coinPriceSource != null) {
            String source = coinPriceSource.getSource();
            try {
                String sourceKey = source + ":ALL_COIN_PRICE";
                if (redisHandler.hasKey(sourceKey)) {
                    Map<Object, Object> cmcCoinPriceMap = redisHandler.hmGetAll(sourceKey);
                    if (cmcCoinPriceMap != null) {
                        List<RelativeCoinPrice> relativeCoinPrices = new ArrayList<>(redisHandler.hLen(sourceKey).intValue());
                        cmcCoinPriceMap.forEach((key, value) -> {
                            RelativeCoinPrice relativeCoinPrice = new RelativeCoinPrice();
                            relativeCoinPrice.setCoinName(String.valueOf(key));
                            relativeCoinPrice.setCoinPrice(new BigDecimal(String.valueOf(value)).setScale(2, BigDecimal.ROUND_HALF_UP).toPlainString());
                            relativeCoinPrices.add(relativeCoinPrice);
                        });
                        return relativeCoinPrices;
                    }
                }
                // 优化点，针对大数据量，下面可以定义几个线程并行处理，提高速度
                List<Future<List<RelativeCoinPrice>>> futures = new ArrayList<>(CORE_THREADS);
                Callable<List<RelativeCoinPrice>> callableTask;
                int subIndex = CMC_RESULT_SIZE_LIMIT / CORE_THREADS;
                String url = coinPriceSource.getPlatformUrl() + URL_SUFFIX;
                for (int i = 0; i < CORE_THREADS; i++) {
                    int startIndex = subIndex * i;
                    // int endOffset = subIndex * (i + 1) - startIndex;
                    callableTask = () -> {
                        List<NameValuePair> paratmers = new ArrayList<>();
                        paratmers.add(new BasicNameValuePair("start", String.valueOf(startIndex == 0 ? 1 : startIndex)));
                        paratmers.add(new BasicNameValuePair("limit", String.valueOf(subIndex)));
                        paratmers.add(new BasicNameValuePair("convert", "USD"));
                        String result = makeAPICall(url, paratmers);
                        List<RelativeCoinPrice> relativeCoinPriceList = new ArrayList<>();

                        if (!StringUtils.isEmpty(result)) {
                            JSONObject coinPriceJson = JSONObject.parseObject(result);
                            if (coinPriceJson.containsKey("status")) {
                                JSONObject statusJson = coinPriceJson.getJSONObject("status");
                                if (null != statusJson) {
                                    Integer errorCode = statusJson.getInteger("error_code");
                                    if (errorCode == 0 && coinPriceJson.containsKey("data")) {
                                        JSONArray dataJsonArray = coinPriceJson.getJSONArray("data");
                                        if (null != dataJsonArray && dataJsonArray.size() > 0) {
                                            RelativeCoinPrice relativeCoinPrice;
                                            JSONObject dataJson;
                                            JSONObject quoteJson;
                                            JSONObject usdJson;
                                            String coinName;
                                            String price;
                                            String dictKeyByDictValue;
                                            for (Object dataJsonObject : dataJsonArray) {
                                                dataJson = JSON.parseObject(JSON.toJSONString(dataJsonObject));
                                                if (null != dataJson) {
                                                    quoteJson = dataJson.getJSONObject("quote");
                                                    if (quoteJson != null) {
                                                        usdJson = quoteJson.getJSONObject("USD");
                                                        if (usdJson != null) {
                                                            relativeCoinPrice = new RelativeCoinPrice();
                                                            coinName = dataJson.getString("symbol");
                                                            relativeCoinPrice.setCoinName(coinName);
                                                            price = usdJson.getString("price");
                                                            relativeCoinPrice.setCoinPrice(new BigDecimal(price).setScale(2, BigDecimal.ROUND_HALF_UP).toPlainString());
                                                            relativeCoinPriceList.add(relativeCoinPrice);
                                                            dictKeyByDictValue = coinPriceDictService.getDictValueByDictKey(source);
                                                            if (StringUtils.isEmpty(dictKeyByDictValue)) {
                                                                dictKeyByDictValue = "60";
                                                            }
                                                            redisHandler.hsetIfAbsent(sourceKey, coinName, price, Long.parseLong(dictKeyByDictValue));
                                                        }

                                                    }
                                                }

                                            }
                                        }

                                    }

                                }

                            }
                        }
                        return relativeCoinPriceList;
                    };
                    futures.add(coinMarketCapThreadPool.submit(callableTask));
                }

                log.info("totalCoinPriceFutures size:{}", futures.size());
                List<RelativeCoinPrice> totalCoinPriceList = new ArrayList<>();
                for (Future<List<RelativeCoinPrice>> future : futures) {
                    totalCoinPriceList.addAll(future.get());
                }
                log.info("totalCoinPriceList size:{}", totalCoinPriceList.size());
                return totalCoinPriceList;

            } catch (Exception e) {
                log.error("invoke cmc to get all coin price error: ", e);
                CoinPriceSource nextCoinPriceSource = coinPriceSourceRouteService.getNextCoinPriceSourceBySourceOnInvokeApiError(source);
                if (nextCoinPriceSource != null) {
                    return coinPriceInvokeHandler.getCoinPriceList(nextCoinPriceSource);
                }
            }
        }
        return Collections.emptyList();
    }

    @Override
    public String getCoinPrice(String source, String coinType) {
        try {
            String sourceKey = source + ":SINGLETON_COIN_PRICE";
            String coinName = coinType.toUpperCase();
            String coinPrice = (String) redisHandler.hget(sourceKey, coinName);
            if (!StringUtils.isEmpty(coinPrice)) {
                log.info("coinName :{} coin price in redis:{} for CMC", coinName, coinPrice);
                return coinPrice;
            }
            CoinPriceSource coinPriceSource = coinPriceSourceService.getCoinPriceSourceBySource(source);
            if (null != coinPriceSource) {
                String url = coinPriceSource.getPlatformUrl() + "/v1/cryptocurrency/quotes/latest";
                List<NameValuePair> paratmers = new ArrayList<>();
                paratmers.add(new BasicNameValuePair("symbol", coinType));
                String result = makeAPICall(url, paratmers);
                if (!StringUtils.isEmpty(result)) {
                    JSONObject coinPriceJson = JSONObject.parseObject(result);
                    if (coinPriceJson.containsKey("status")) {
                        JSONObject statusJson = coinPriceJson.getJSONObject("status");
                        if (null != statusJson) {
                            Integer errorCode = statusJson.getInteger("error_code");
                            if (errorCode == 0 && coinPriceJson.containsKey("data")) {
                                JSONObject dataJson = coinPriceJson.getJSONObject("data");
                                if (null != dataJson) {
                                    if (dataJson.containsKey(coinName)) {
                                        JSONObject coinTypeJson = dataJson.getJSONObject(coinName);
                                        if (coinTypeJson != null) {
                                            JSONObject quoteJson = coinTypeJson.getJSONObject("quote");
                                            if (quoteJson != null) {
                                                JSONObject usdJson = quoteJson.getJSONObject("USD");
                                                if (usdJson != null) {
                                                    coinPrice = new BigDecimal(usdJson.getString("price")).setScale(2, BigDecimal.ROUND_HALF_UP).toPlainString();
                                                    String dictKeyByDictValue = coinPriceDictService.getDictValueByDictKey(source);
                                                    if (StringUtils.isEmpty(dictKeyByDictValue)) {
                                                        dictKeyByDictValue = "60";
                                                    }
                                                    redisHandler.hset(sourceKey, coinName, coinPrice, Integer.parseInt(dictKeyByDictValue));
                                                    return coinPrice;
                                                }

                                            }
                                        }
                                    }
                                }

                            }

                        }

                    }
                }
            }
        } catch (Exception e) {
            log.error("invoke cmc to get singleton coin price error: ", e);
            CoinPriceSource nextCoinPriceSource = coinPriceSourceRouteService.getNextCoinPriceSourceBySourceOnInvokeApiError(source);
            if (nextCoinPriceSource != null) {
                return coinPriceInvokeHandler.getCoinPrice(nextCoinPriceSource.getSource(), coinType);
            }
        }
        return CommonConstant.DEFAULT_COIN_PRICE;
    }

    @PreDestroy
    public void destroy() {
        coinMarketCapThreadPool.shutdown();
    }

    private String makeAPICall(String uri, List<NameValuePair> parameters)
            throws URISyntaxException, IOException {

        URIBuilder query = new URIBuilder(uri);
        query.addParameters(parameters);

        CloseableHttpClient client = HttpClients.createDefault();
        HttpGet request = new HttpGet(query.build());

        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectTimeout(connectTimeOut)
                .setConnectionRequestTimeout(connectRequestTimeout)
                .setSocketTimeout(socketTimeOut).build();
        request.setConfig(requestConfig);

        request.setHeader(HttpHeaders.ACCEPT, "application/json");
        request.setHeader(HttpHeaders.ACCEPT_ENCODING, "gzip,deflate");
        request.addHeader("X-CMC_PRO_API_KEY", coinPriceDictService.getDictValueByDictKey(CommonConstant.CMC_API_KEY));

        CloseableHttpResponse response = client.execute(request);

        String response_content;
        try {
            // System.out.println(response.getStatusLine());
            HttpEntity entity = response.getEntity();
            response_content = EntityUtils.toString(entity);
            EntityUtils.consume(entity);
        } finally {
            response.close();
        }

        return response_content;
    }

}
